//==========================================================================
// Copyright (c) 2000-2008,  Elastos, Inc.  All Rights Reserved.
//=========================================================================
#include <winhack.h>
#include <elasys_server.h>
#include "eventqueue.h"
#include "linknode.h"

CEventQueue::~CEventQueue()
{
    if (NULL != m_hEvent) {
        CloseHandle(m_hEvent);
        m_hEvent = NULL;
    }
}

ECode CEventQueue::Initialize()
{
    m_hEvent = CreateEventW(NULL, FALSE, FALSE, NULL);
    if (NULL == m_hEvent) {
         // todo ERROR;
         return E_FAIL;
    }

    // AddNotify
    return NOERROR;
}

ECode CEventQueue::Read(
    /* [in] */ Int64 offset,
    /* [in] */ Int32 unumberOfBytesToRead,
    /* [out] */ MemoryBuf * pData,
    /* [out] */ IEvent * * ppCompletionEvent)
{
    Boolean bAsync = (offset & 0x80000000) ? TRUE : FALSE;

    if ((NULL == pData)
        || (pData->GetCapacity() < m_queue.ELEM_SIZ)) {
        return E_INVALID_ARGUMENT;
    }

    int ret = -1;
    while (0 != ret) {
        ret = m_queue.Dequeue((char*)pData->GetPayload());
        if (0 == ret) {
            pData->SetUsed(m_queue.ELEM_SIZ);
            break;
        }
        if (bAsync) {//no data
            return E_INVALID_OPERATION;//E_NOT_READY
        }
        WaitForSingleObjectEx(m_hEvent, (DWORD)INFINITE, FALSE);
    }

    return NOERROR;
}

ECode CEventQueue::Write(
    /* [in] */ Int64 offset,
    /* [in] */ const MemoryBuf & ebbData,
    /* [out] */ Int32 * pNumberOfBytesWritten,
    /* [out] */ IEvent * * ppCompletionEvent)
{
    void * param = (void *)ebbData.GetPayload();

    m_queue.Enqueue(param);
    SetEvent(m_hEvent);

    return NOERROR;
}

ECode CEventQueue::Control(
    /* [in] */ Handle32 controlCode,
    /* [in] */ const MemoryBuf & ebbInData,
    /* [out] */ MemoryBuf * pOutData,
    /* [out] */ IEvent * * ppCompletionEvent)
{
    // TODO;
    return E_NOT_IMPLEMENTED;
}

void CEventQueue::Dispose()
{
   delete this;
}

// -------------------------------------------------------------


static CEventQueueSink g_eventQueueSink;

CEventQueueSink::CEventQueueSink()
{
    for (Int32 i = 0; i < HOST_NUMBER; i++) {
        m_queueList[i].Initialize();
    }
}

ECode CEventQueueSink::CreateQueue(
    /* [in] */ Int32 environmentId,
    /* [in] */ Int32 hostType,
    /* [in] */ Int32 elmentSize,
    /* [in] */ Int32 queueSize,
    /* [out] */ CEventQueue ** ppQueue)
{
    CEventQueue * pQueue = NULL;
    ECode ec = NOERROR;

    assert(hostType < HOST_NUMBER);

    pQueue = new CEventQueue(environmentId, elmentSize, queueSize);
    if (NULL == pQueue) {
        ec = E_OUT_OF_MEMORY;
        goto ErrorExit;
    }
    ec = pQueue->Initialize();
    if (FAILED(ec)) {
        goto ErrorExit;
    }
    pQueue->AddRef();

   // Lock;
    m_queueList[hostType].InsertLast(pQueue);
   // Unlock;

    *ppQueue = pQueue;
    return NOERROR;

ErrorExit:
    if (NULL != pQueue) {
        delete pQueue;
        pQueue = NULL;
    }
    return ec;
}

CEventQueue * CEventQueueSink::GetQueue(
        /* [in] */ Int32 enviromentId,
        /* [in] */ Int32 hostType)
{
    assert(hostType < HOST_NUMBER);
    assert(!m_queueList[hostType].IsEmpty());

    CEventQueue * pQueue = (CEventQueue *)m_queueList[hostType].Next();
    assert(enviromentId == pQueue->m_environmentId);

    pQueue->AddRef();
    return pQueue;
}

CEventQueueSink * GetEventQueueSink()
{
    return &g_eventQueueSink;
}

